package com.lwl.boot;

import com.lwl.client.RPCConsumer;
import com.lwl.handler.ZkNodeWatcher;
import com.lwl.handler.ZookeeperWatcher;
import com.lwl.pojo.RpcConnectPojo;
import com.lwl.service.IUserService;
import com.lwl.utils.Constants;
import com.lwl.utils.DateUtils;
import com.lwl.zk.ZookeeperUtils;
import org.I0Itec.zkclient.ZkClient;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConsumerBoot {

    static ZkClient session = ZookeeperUtils.getSession();
    static CountDownLatch countDownLatch = new CountDownLatch(1);
    static ExecutorService cachedThreadPool = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws InterruptedException {
        System.setProperty("io.netty.noUnsafe", "true");

        //获取根节点下的子节点
        List<String> children = session.getChildren(Constants.ZK_BASEPATH);

        //作业一
        //启动客户端，并添加监听
        connectedServer(children);
        ZkNodeWatcher.setZkNodeWatcher();
        //循环给服务端发送信息
        new Thread(() -> sendMsgToServer()).start();

        //作业二
        sendMsgToServer2();
        countDownLatch.await();
    }

    //启动客户端，并添加监听
    public static void connectedServer(List<String> children) throws InterruptedException {
        Set<String> hashSet = new HashSet<>();
        for (String child : children) {
            String path = Constants.ZK_BASEPATH+"/" + child;
            if (!ZookeeperWatcher.serviceMap.containsKey(path)) {
                hashSet.add(path);
                String[] ipPort = child.split(":");
                RPCConsumer rpcConsumer = new RPCConsumer(ipPort[0], ipPort[1]);
                //创建代理对象
                IUserService service = (IUserService) rpcConsumer.createProxy(IUserService.class);
                RpcConnectPojo rpcConnectPojo = new RpcConnectPojo();
                rpcConnectPojo.setUserService(service);
                rpcConnectPojo.setEventLoopGroup(rpcConsumer.group);

                ZookeeperWatcher.serviceMap.put(path, rpcConnectPojo);
                System.out.println(child + " 客户端和服务端建立连接");
            }
        }

        //设置服务端节点监听
        if (hashSet.size() > 0) {
            ZookeeperWatcher.setWatcher(hashSet);
        }
    }

    //循环给服务端连接并发送测试数据
    public static void sendMsgToServer() {
        // 循环给服务器写数据
        while (true) {
            try {
//                Iterator<String> iterator = ZookeeperUtils.serviceMap.keySet().iterator();
//                while (iterator.hasNext()) {
                for (String key : ZookeeperWatcher.serviceMap.keySet()) {
                    cachedThreadPool.execute(() -> {
                        try {
//                            String key = iterator.next();
                            RpcConnectPojo service = ZookeeperWatcher.serviceMap.get(key);
                            String msg = key + " >>> " + DateUtils.getDataString(DateUtils.datetimeFormat);

                            System.out.println("客户端发送信息：" + msg);
                            String result = service.getUserService().sayHello(msg);
                            System.out.println("客户端接收服务端信息：" + key + "  " + result);
                        } catch (Exception e) {
                            System.out.println(e);
                        }
                    });
                }
                int random = new Random().nextInt(10);
                //随机睡眠
                System.out.println("随机睡眠" + random);
                Thread.sleep(random * 1000);

            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

    //负载均衡给服务端连接并发送测试数据
    public static void sendMsgToServer2() {
        // 循环给服务器写数据
        while (true) {
            try {
                //节点最后一次响应的节点
                Map<String, Long> map = new HashMap<>();
                //获取各节点的时间
                ZkClient session = ZookeeperUtils.getSession();
                List<String> children = session.getChildren(Constants.ZK_BASEPATH);
                for (String child : children) {
                    long nowTime = new Date().getTime();
                    String path = Constants.ZK_BASEPATH+ "/" + child;
                    Long lastDate = session.readData(path);
                    lastDate = lastDate == null ? 0 : lastDate;
                    Long difTime = nowTime - lastDate;
                    //如果超过5秒赋值0
                    if (difTime > 5 * 1000) {
                        ZookeeperUtils.writeData(path, 0L);
                        map.put(path, 0L);
                    } else {
                        map.put(path, difTime);
                    }
                }
                for (String key : map.keySet()) {
                    Long aLong = map.get(key);
                    System.out.println("节点：" + key + "  最后响应时间：" + aLong);
                }

                Optional<Map.Entry<String, Long>> minEntry = map.entrySet().stream().filter(x -> x.getValue() > 0).min(Comparator.comparingLong(x -> x.getValue()));

                //如果都超过5秒则默认取第一个
                if (minEntry.isEmpty()) {
                    minEntry = map.entrySet().stream().findFirst();
                }
                //最后响应的路径
                String path = minEntry.get().getKey();
                if (ZookeeperWatcher.serviceMap.containsKey(path)) {
                    RpcConnectPojo rpcConnectPojo = ZookeeperWatcher.serviceMap.get(path);
                    try {
                        String msg = ">>>>>>>>>> 负载测试 " + path + " " + DateUtils.getDataString(DateUtils.datetimeFormat);
                        System.out.println("负载测试>>>>>>>>>>客户端发送信息：" + msg);
                        String result = rpcConnectPojo.getUserService().sayHello(msg);
                        System.out.println("负载测试>>>>>>>>>>客户端接收服务端信息：" + result);
                    } catch (Exception e) {
                        System.out.println("负载测试>>>>>>>>>>" + e);
                    }
                }
                Thread.sleep(2000);
            } catch (Exception e) {
                System.out.println(e);
            }
        }
    }

}